package cn.fw.community.infrastructure.kafka;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.Optional;
import java.util.function.Consumer;

/**
 * @author cn
 * @version 1.0
 * @description
 * @date 2022/10/8 13:45
 */

@Component
@Slf4j
@RequiredArgsConstructor
public class KafKaProducer {
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private static final String EVENT_TOPIC = "event_tracking";

    public void sendEventTrackingMessage(Object msg, Consumer<SendResult<String, Object>> successCb, Consumer<Throwable> failCb) {
        ListenableFuture<SendResult<String, Object>> send = kafkaTemplate.send(EVENT_TOPIC, msg);
        send.addCallback(success -> {
            if (successCb != null) {
                successCb.accept(success);
            }
            Optional.ofNullable(success).map(SendResult::getRecordMetadata).ifPresent(e -> log.info("kafka发送成功:topic:{};partition:{};offset:{};msg:{}", e.topic(), e.partition(), e.offset(), msg));
        }, fail -> {
            if (failCb != null) {
                failCb.accept(fail);
            }
            log.info("kafka发送失败:topic:{};msg:{}", EVENT_TOPIC, fail.getMessage());
        });
    }

    public void sendEventTrackingMessage(Object msg) {
        sendEventTrackingMessage(msg, null, null);
    }
}
